#include "messaging/kafka_producer.h"
#include "messaging/kafka_consumer.h"
#include "common/logger/logger.h"
#include <iostream>
#include <thread>
#include <chrono>
#include <atomic>
#include <csignal>

using namespace messaging;
using namespace common::logger;

// 全局变量，用于信号处理
std::atomic<bool> running(true);

// 信号处理函数
void signalHandler(int signum) {
    std::cout << "收到信号: " << signum << ", 准备退出..." << std::endl;
    running = false;
}

int main(int argc, char* argv[]) {
    // 注册信号处理
    signal(SIGINT, signalHandler);
    signal(SIGTERM, signalHandler);
    
    // 初始化日志
    Logger::getInstance().setLevel(LogLevel::INFO);
    Logger::getInstance().addSink(std::make_shared<ConsoleSink>());
    
    // 默认参数
    std::string mode = "producer";
    std::string topic = "test";
    std::string message = "Hello Kafka!";
    
    // 检查参数
    if (argc > 1) {
        mode = argv[1];
    } else {
        std::cout << "未指定模式，默认使用: " << mode << std::endl;
        std::cout << "可用模式: producer, consumer" << std::endl;
    }
    
    if (argc > 2) {
        topic = argv[2];
    } else {
        std::cout << "未指定主题，默认使用: " << topic << std::endl;
    }
    
    if (argc > 3) {
        message = argv[3];
    } else if (mode == "producer") {
        std::cout << "未指定消息内容，默认使用: " << message << std::endl;
    }
    
    std::cout << "运行模式: " << mode << ", 主题: " << topic << std::endl;
    
    try {
        if (mode == "producer") {
            // 创建生产者配置
            KafkaProducerConfig config;
            config.brokers = "kafka:9092"; // 修改为本地地址
            config.clientId = "example-producer";
            
            // 创建生产者
            auto producer = KafkaProducer::create(config);
            if (!producer) {
                LOG_ERROR("创建Kafka生产者失败");
                return 1;
            }
            
            LOG_INFO("Kafka生产者已创建");
            
            // 发送消息
            int count = 0;
            while (running && count < 10) {
                std::string key = "key-" + std::to_string(count);
                std::string value = message + " #" + std::to_string(count);
                
                LOG_INFO("发送消息: " + value);
                
                bool success = producer->send(topic, key, value, [key, value](bool success, const std::string& error) {
                    if (success) {
                        LOG_INFO("消息发送成功: " + key + " -> " + value);
                    } else {
                        LOG_ERROR("消息发送失败: " + key + ", 错误: " + error);
                    }
                });
                
                if (!success) {
                    LOG_ERROR("发送消息失败");
                }
                
                count++;
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
            
            // 刷新所有未发送的消息
            producer->flush();
            LOG_INFO("生产者已退出");
        } else if (mode == "consumer") {
            // 创建消费者配置
            KafkaConsumerConfig config;
            config.brokers = "kafka:9092"; // 修改为本地地址
            config.groupId = "example-group";
            config.clientId = "example-consumer";
            config.offsetReset = "earliest";  // 从最早的消息开始消费
            
            // 创建消费者
            auto consumer = KafkaConsumer::create(config);
            if (!consumer) {
                LOG_ERROR("创建Kafka消费者失败");
                return 1;
            }
            
            LOG_INFO("Kafka消费者已创建");
            
            // 订阅主题
            if (!consumer->subscribe({topic})) {
                LOG_ERROR("订阅主题失败: " + topic);
                return 1;
            }
            
            LOG_INFO("已订阅主题: " + topic + ", 等待消息...");
            
            // 启动消费
            if (!consumer->start(
                [](const std::string& topic, const std::string& key, const std::string& value, int64_t offset) {
                    LOG_INFO("收到消息: Topic=" + topic + ", Key=" + key + ", Value=" + value +
                           ", Offset=" + std::to_string(offset));
                },
                [](const std::string& error) {
                    LOG_ERROR("Kafka错误: " + error);
                })) {
                LOG_ERROR("启动消费者失败");
                return 1;
            }
            
            // 运行一段时间
            while (running) {
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
            
            // 停止消费
            consumer->stop();
            LOG_INFO("消费者已退出");
        } else {
            std::cerr << "无效的模式: " << mode << ", 应为 producer 或 consumer" << std::endl;
            return 1;
        }
    } catch (const std::exception& e) {
        LOG_ERROR("异常: " + std::string(e.what()));
        return 1;
    }
    
    return 0;
} 